package com.sub.framework.config;

import com.sub.framework.rabbitmq.message.ProtoMessageConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

//@Configuration
public class RabbitMQConfig {

     /** 定义日志*/
     private static  final Logger loger = LoggerFactory.getLogger(RabbitMQConfig.class);
     /** 自动配置RabbitMQ 的连接工厂*/
     @Autowired
    private CachingConnectionFactory cachingConnectionFactory;
    //自动装配消息监听所在的容器工厂配置类实例
     @Autowired
     private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigure;

    /**
     *  单一消费者配置实例
      * @return
     */
//     @Bean(name = "singleListenerContainer")
     public SimpleRabbitListenerContainerFactory listenerContainerFactory(){
         //定义消息监听所在的容器工厂
         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
         /** 设置容器工厂所用的实例*/
         factory.setConnectionFactory(cachingConnectionFactory);
         /** 设置消息在传输中的格式，这里采用JSON 的格式进行传输*/
         factory.setMessageConverter(new ProtoMessageConverter());
        /** 设置并发消费者的初始化数量，这里设置为一个*/
        factory.setConcurrentConsumers(1);
        /** 设置消费者最大并发数量.,这里设置1个*/
        factory.setMaxConcurrentConsumers(1);
       /** 设置并发消费者实例中每个实例拉取的消息数量*/
       factory.setPrefetchCount(1);
        return factory;
     }
    /**
     *  单一消费者配置实例 确认模式为AUTO
     * @return
     */
//    @Bean(name = "singleListenerContainerAuto")
     public SimpleRabbitListenerContainerFactory listenerContainerFactoryAuto(){
         //定义消息监听所在的容器工厂
         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
         /** 设置容器工厂所用的实例*/
         factory.setConnectionFactory(cachingConnectionFactory);
         /** 设置消息在传输中的格式，这里采用JSON 的格式进行传输*/
//         factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setMessageConverter(new ProtoMessageConverter());

        /** 设置并发消费者的初始化数量，这里设置为一个*/
         factory.setConcurrentConsumers(1);
         /** 设置消费者最大并发数量.,这里设置1个*/
         factory.setMaxConcurrentConsumers(1);
         /** 设置并发消费者实例中每个实例拉取的消息数量*/
         factory.setPrefetchCount(1);
         /** 设置确认消费模式AUTO*/
         factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
         return factory;
     }
    /**
     *  多消费者配置实例
     * @return
     */
//     @Bean(name = "multiListnerContainer")
     public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(){
         //定义消息监听工厂所在的容器
         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
         /** 设置容器工厂所用的实例*/
         factoryConfigure.configure(factory,cachingConnectionFactory);
         /** 设置消息格式的传输格式，*/
         /** 设置消息在传输中的格式，这里采用JSON 的格式进行传输*/
         factory.setMessageConverter(new Jackson2JsonMessageConverter());
         /** 设置消息的确认模式。这里为NONE ,表示不需要确认消费*/
         factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
         /** 设置并发消费者的初始化数量，这里设置为一个*/
         factory.setConcurrentConsumers(10);
         /** 设置消费者最大并发数量.,这里设置5个*/
         factory.setMaxConcurrentConsumers(15);
         /** 设置并发消费者实例中每个实例拉取的消息数量*/
         factory.setPrefetchCount(10);
         return factory;
     }

    /**
     *  自定义配置RabbitMQ 发消息的操作组件RabbitTemplate
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
         // 设置“发送消息后进行确认”
        cachingConnectionFactory.setPublisherConfirms(true);
        /** 设置“发送消息后返回确认信息”*/
        cachingConnectionFactory.setPublisherReturns(true);
        /** 构造发送消息组件实例对象*/
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        rabbitTemplate.setMandatory(true);
//        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMessageConverter(new ProtoMessageConverter());
        // 发消息后如果成功，则输出“消息发送成功”的反馈信息
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                loger.info("消息发送成功：correlationData({}),ack({}),cause({})",correlationData,b,s);
            }
        });

        /** 发送消息后，如果发送失败，则输出“消息发送失败-消息丢失” 的反馈信息*/
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                loger.info("消息丢失：exchange({}),route({}),replyCode({}),replyText({}),message({})",s1,s2,s,message);
            }
        });
        return rabbitTemplate;
    }

    // 定义读取配置文件的环境变量实例
    @Autowired
    private Environment environment;

    /**
     *  c创建简单的消息模型：队列 、 交互机 和路由
     * @return
     */
    @Bean(name = "basicQueue")
    public Queue basicQueue(){
        return  new Queue(environment.getProperty("mq.basic.info.queue.name"),true);
    }

    @Bean
    public TopicExchange basicExchange(){
        return  new TopicExchange(environment.getProperty("mq.basic.info.exchange.name"),true,false);
    }

    @Bean(name = "autoQueue")
    public Queue autoQueue(){
        return  new Queue(environment.getProperty("mq.auto.knowledge.queue.name"),true);
    }

    @Bean
    public TopicExchange autoExchange(){
        return  new TopicExchange(environment.getProperty("mq.auto.knowledge.exchange.name"),true,false);
    }

    /**
     * 创建绑定
     * @return
     */
    @Bean
    public Binding basicBinding() {
        return BindingBuilder
                .bind(basicQueue())
                .to(basicExchange())
                .with(environment.getProperty("mq.basic.info.routing.key.name"));
    }

    @Bean
    public Binding autoBinding() {
        return BindingBuilder
                .bind(autoQueue())
                .to(autoExchange())
                .with(environment.getProperty("mq.auto.knowledge.routing.key.name"));
    }

}
